use std::{collections::VecDeque, sync::Arc, time::Duration};

use bytes::{BufMut, Bytes, BytesMut};
use http::{Method, Version};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    time::timeout,
};

use crate::{
    config::http::{
        core_conf::{HttpCoreRuntimeConf, HEADER_BUF_MIN_WRITEABLE_SIZE},
        main::get_http_main_runtime_conf,
    },
    http::base::{
        conn::BoxHttpConn,
        header_name::HeaderName,
        header_value::HeaderValue,
        headers::{HttpHeader, HttpHeaders},
        parse::{http_read_crlf_line, parse_http_header, HttpResult},
        status_code::{http_body_allow_status, StatusCode},
        Request, Response,
    },
    utils::{fixed_buffer::FixedBuffer, time::get_now_msec},
};

use super::body::{BodyReader, BodyWriter};
use crate::error::*;

#[allow(dead_code)]
pub struct ClientSession {
    hc: BoxHttpConn,

    // conf
    write_header_timeout_ms: usize,
    read_header_timeout_ms: usize,
    body_write_timeout_ms: usize,
    body_read_timeout_ms: usize,
    one_header_buf_size: usize,
    max_header_size: usize,
    ignore_invalid_header: bool,
    body_buf_size: usize,

    want_close: bool,

    // 统计数据
    header_bytes: usize,

    // 存放待解析数据
    last_buf: Option<FixedBuffer>,

    // request
    pub header_send: bool,
    upgrade_req: bool,
    content_length: Option<usize>,
    chunked_transfer: bool,
    expect_write_trailers: usize,
    body_writer: Option<BodyWriter>,
    write_trailers_done: bool,

    // response
    resp: Response,
    resp_no_body: bool,
    parse_rl_done: bool,
    parse_rh_done: bool,
    invalid_headers: usize,
    resp_expect_trailers: usize,
    resp_content_length: Option<usize>,
    resp_chunked_transfer: bool,
    body_reader: Option<BodyReader>,
    finalized: bool,
}

#[allow(dead_code)]
impl ClientSession {
    pub fn new(
        hc: BoxHttpConn,
        conf: Option<Arc<HttpCoreRuntimeConf>>,
    ) -> Self {
        let conf = match conf {
            Some(conf) => conf,
            None => get_http_main_runtime_conf().core_runtime_conf.clone(),
        };

        let write_header_timeout_ms = conf.client_write_header_timeout_ms;
        let read_header_timeout_ms = conf.client_read_header_timeout_ms;
        let body_write_timeout_ms = conf.body_write_timeout_ms;
        let body_read_timeout_ms = conf.body_read_timeout_ms;
        let one_header_buf_size = conf.one_header_buf_size;
        let ignore_invalid_header = conf.ignore_invalid_header;
        let max_header_size = conf.max_header_size;
        let body_buf_size = conf.body_buf_size;

        ClientSession {
            hc,

            // conf
            write_header_timeout_ms,
            read_header_timeout_ms,
            body_write_timeout_ms,
            body_read_timeout_ms,
            one_header_buf_size,
            ignore_invalid_header,
            max_header_size,
            body_buf_size,

            want_close: false,

            // 统计数据
            header_bytes: 0,

            // 存放待解析数据
            last_buf: None,

            // request
            header_send: false,
            upgrade_req: false,
            content_length: None,
            chunked_transfer: false,
            expect_write_trailers: 0,
            body_writer: None,
            write_trailers_done: false,

            // response
            resp: Response::new(),
            resp_no_body: false,
            parse_rl_done: false,
            parse_rh_done: false,
            invalid_headers: 0,
            resp_expect_trailers: 0,
            resp_content_length: None,
            resp_chunked_transfer: false,
            body_reader: None,

            finalized: false,
        }
    }

    // ClientSession不保存Request header.
    pub async fn send_req_header(&mut self, req: &Request) -> Result<()> {
        if self.header_send {
            return Error::e_explain(
                ErrorType::InternalError,
                "request header already send done",
            );
        }
        self.header_send = true;

        self.hc.requests += 1;

        if req.method == Method::HEAD {
            self.resp_no_body = true;
        }

        let method = req.method.as_str();

        let mut http_version = b"HTTP/1.1";
        self.resp.version = Version::HTTP_11;

        if req.version == Version::HTTP_10 {
            http_version = b"HTTP/1.0";
            self.want_close = true;
            self.resp.version = Version::HTTP_10;
        }

        let request_uri = req
            .uri
            .path_and_query()
            .if_none(ErrorType::InvalidRequestLine, "invalid request uri")?
            .as_str();

        // Eg: Method Path?args HTTP/1.1\r\n
        let mut len =
            method.len() + 1 + request_uri.len() + 1 + http_version.len() + 2;

        for v in req.headers.iter() {
            let header_name = v.get_name();
            match header_name.len() {
                7 => {
                    if header_name.eq_ignore_ascii_case(HeaderName::TRAILER) {
                        let iter = v.get_value().split(|c| *c == b',');
                        for mut v in iter {
                            v = v.trim_ascii();
                            if !v.is_empty() {
                                self.expect_write_trailers += 1;
                            }
                        }
                    }

                    if header_name.eq_ignore_ascii_case(HeaderName::UPGRADE) {
                        self.upgrade_req = true;
                    }
                }

                10 => {
                    if header_name.eq_ignore_ascii_case(HeaderName::CONNECTION)
                        && v.value_contain(b"close")
                    {
                        self.want_close = true;
                    }
                }

                14 => {
                    if header_name
                        .eq_ignore_ascii_case(HeaderName::CONTENT_LENGTH)
                    {
                        let res = String::from_utf8_lossy(v.get_value())
                            .to_string()
                            .parse()
                            .if_err(
                                ErrorType::InvalidHttpHeader,
                                "parse content-length header failed",
                            )?;
                        self.content_length = Some(res);
                    }
                }

                17 => {
                    if header_name
                        .eq_ignore_ascii_case(HeaderName::TRANSFER_ENCODING)
                        && v.value_contain(HeaderValue::CHUNKED)
                    {
                        self.chunked_transfer = true;
                    }
                }

                _ => {
                    // nothing
                }
            }

            // Name: Value\r\n
            len += v.get_name().len() + 2 + v.get_value().len() + 2;
        }

        let mut header_buf = BytesMut::with_capacity(len);
        header_buf.put_slice(method.as_bytes());
        header_buf.put_bytes(b' ', 1);
        header_buf.put_slice(request_uri.as_bytes());
        header_buf.put_bytes(b' ', 1);
        header_buf.put_slice(http_version);
        header_buf.put_slice(b"\r\n");

        for v in req.headers.iter() {
            // Name: Value\r\n
            header_buf.put_slice(v.get_name());
            header_buf.put_slice(b": ");
            header_buf.put_slice(v.get_value());
            header_buf.put_slice(b"\r\n");
        }
        header_buf.put_slice(b"\r\n");

        let write_event = self.hc.write_all(&header_buf);
        let write_event = timeout(
            Duration::from_millis(self.write_header_timeout_ms as u64),
            write_event,
        );

        write_event
            .await
            .if_err(ErrorType::WriteTimedout, "send rqeuest header timeout")?
            .if_err(ErrorType::WriteError, "send request header error")?;

        self.init_body_writer();

        Ok(())
    }

    pub async fn write_req_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
        done: bool,
    ) -> Result<HttpResult> {
        match &mut self.body_writer {
            Some(writer) => writer.write_body(&mut self.hc, bufs, done).await,
            None => Error::e_explain(
                ErrorType::InternalError,
                "body_writer no inited",
            ),
        }
    }

    pub fn expect_write_trailers_num(&mut self) -> usize {
        self.expect_write_trailers
    }

    pub async fn write_req_trailers(
        &mut self,
        headers: &HttpHeaders,
    ) -> Result<()> {
        if !self.write_request_body_done() {
            return Error::e_explain(
                ErrorType::InternalError,
                "request body has not been sent completely",
            );
        }

        if self.write_trailers_done {
            return Error::e_explain(
                ErrorType::InternalError,
                "request trailers already write",
            );
        }

        // 某些传输场景不支持trailers. 此处我们直接返回Ok.
        if self.expect_write_trailers == 0 {
            self.write_trailers_done = true;
            return Ok(());
        }

        self.body_writer
            .as_mut()
            .unwrap()
            .write_trailers(&mut self.hc, headers)
            .await?;
        self.write_trailers_done = true;

        Ok(())
    }

    fn write_request_body_done(&self) -> bool {
        match &self.body_writer {
            Some(writer) => writer.done(),
            None => false,
        }
    }

    pub fn write_request_done(&self) -> bool {
        if !self.write_request_body_done() {
            return false;
        }

        // write_request_done要额外判断trailers是否发送.
        if self.expect_write_trailers > 0 && !self.write_trailers_done {
            return false;
        }

        true
    }

    pub async fn read_response(&mut self) -> Result<&mut Response> {
        self.read_response_inner().await?;

        // 服务端可能响应非101的其他状态码.
        // 此时我们需要清空相关状态, 并继续读取response header.
        if self.resp.status_code >= 100
            && self.resp.status_code <= 199
            && self.resp.status_code != StatusCode::HTTP_SWITCHING_PROTOCOLS
        {
            self.resp.clear();
            self.parse_rl_done = false;
            self.parse_rh_done = false;
            self.resp_expect_trailers = 0;
            self.resp_content_length = None;
            self.resp_chunked_transfer = false;
            self.read_response_inner().await?;
        }

        self.process_response()?;
        Ok(&mut self.resp)
    }

    pub async fn read_resp_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
    ) -> Result<HttpResult> {
        match &mut self.body_reader {
            Some(reader) => reader.read_body(&mut self.hc, bufs).await,
            None => Error::e_explain(
                ErrorType::InternalError,
                "body reader no inited",
            ),
        }
    }

    pub fn read_resp_done(&mut self) -> bool {
        match &self.body_reader {
            Some(reader) => reader.done(),
            None => false,
        }
    }

    pub fn expect_trailers_num(&self) -> usize {
        self.resp_expect_trailers
    }

    // 在read_resp_body完毕后方可调用.
    pub fn get_resp_trailer_headers(&self) -> Result<Option<&HttpHeaders>> {
        match &self.body_reader {
            Some(reader) => reader.get_trailers(),
            None => Error::e_explain(
                ErrorType::InternalError,
                "body reader no inited",
            ),
        }
    }

    // 如果该连接可以复用, 返回hc.
    pub async fn finalize(&mut self, abort: bool) {
        if self.finalized {
            return;
        }
        self.finalized = true;

        if abort {
            self.want_close = true;
        }

        if !self.write_request_done() {
            self.want_close = true;
        }

        if !self.read_resp_done() {
            self.want_close = true;
        }

        let body_reader = self.body_reader.take();
        if let Some(reader) = body_reader {
            // 有残余数据, 不可复用.
            let bytes = reader.get_remaining_data();
            if !bytes.is_empty() {
                self.want_close = true;
            }
        }

        if self.want_close {
            _ = self.hc.shutdown().await;
        }
    }

    pub fn release(self) -> Option<BoxHttpConn> {
        if !self.finalized {
            log::error!("expect client session finalized, but got false");
            return None;
        }

        if self.want_close {
            return None;
        }

        Some(self.hc)
    }
}

impl ClientSession {
    fn init_body_writer(&mut self) {
        if self.upgrade_req {
            self.expect_write_trailers = 0;
            self.want_close = true;
            self.body_writer = Some(BodyWriter::with_until_closed_transfer(
                self.body_write_timeout_ms,
            ));
            return;
        }

        if self.chunked_transfer {
            self.body_writer = Some(BodyWriter::with_chunked_transfer(
                self.body_write_timeout_ms,
                self.expect_write_trailers,
            ));
            return;
        }

        if let Some(cl) = self.content_length {
            self.expect_write_trailers = 0;
            self.body_writer = Some(BodyWriter::with_cl_transfer(
                cl,
                self.body_write_timeout_ms,
            ));
            return;
        }

        self.expect_write_trailers = 0;
        self.body_writer = Some(BodyWriter::with_no_body());
    }

    async fn read_response_inner(&mut self) -> Result<&mut Response> {
        let mut next_read_timeout = false;
        let mut read_header_timeout =
            Duration::from_millis(self.read_header_timeout_ms as u64);

        self.last_buf = Some(FixedBuffer::new(self.one_header_buf_size));

        loop {
            let res = self.parse_response()?;
            // parse response header success.
            if res == HttpResult::Complete {
                return Ok(&mut self.resp);
            }

            if self.header_bytes > self.max_header_size {
                return Error::e_explain(
                    ErrorType::HttpHeaderLarge,
                    "response header is too large",
                );
            }

            if next_read_timeout {
                return Error::e_explain(
                    ErrorType::ReadTimedout,
                    "read response header timedout",
                );
            }

            let last_buf = self.last_buf.as_mut().unwrap();

            // 写空间不足, 压实一下buf.
            if last_buf.write_slice_len() < HEADER_BUF_MIN_WRITEABLE_SIZE {
                last_buf.compaction();
            }

            let write_slice = last_buf.write_slice();
            let read_event = self.hc.read(write_slice);
            let read_event = timeout(read_header_timeout, read_event);

            let start_ms = get_now_msec();

            let res = read_event
                .await
                .if_err(
                    ErrorType::ReadTimedout,
                    "read response header timedout",
                )?
                .if_err(ErrorType::ReadError, "read response header error")?;

            let interval = get_now_msec() - start_ms;

            // 低于3ms, 下次读取前直接返回Timedout.
            read_header_timeout = read_header_timeout
                .saturating_sub(Duration::from_millis(interval));
            if read_header_timeout.as_millis() < 3 {
                next_read_timeout = true;
            }

            // Read EOF.
            if res == 0 {
                return Error::e_explain(
                    ErrorType::ReadEOF,
                    "upstream closed in read response header process",
                );
            }

            last_buf.fill(res);
        }
    }

    fn parse_response(&mut self) -> Result<HttpResult> {
        loop {
            let line_len = {
                let read_slice = self.last_buf.as_ref().unwrap().read_slice();
                match http_read_crlf_line(read_slice) {
                    Some(n) => n,
                    None => {
                        // big header line, return error.
                        if read_slice.len()
                            > self.one_header_buf_size
                                - (2 * HEADER_BUF_MIN_WRITEABLE_SIZE)
                        {
                            return Error::e_explain(
                                ErrorType::BigHeaderLine,
                                format!(
                                    "found big header line, length {} more than {}",
                                    read_slice.len(),
                                    self.one_header_buf_size - (2 * HEADER_BUF_MIN_WRITEABLE_SIZE)
                                ),
                            );
                        }

                        // no more data
                        return Ok(HttpResult::Partial);
                    }
                }
            };

            self.parse_response_line(line_len, self.ignore_invalid_header)?;

            let last_buf = self.last_buf.as_mut().unwrap();
            self.header_bytes += line_len;
            last_buf.consume(line_len);

            if self.parse_rh_done {
                return Ok(HttpResult::Complete);
            }
        }
    }

    fn parse_response_line(
        &mut self,
        line_len: usize,
        ignore_invalid_header: bool,
    ) -> Result<()> {
        if !self.parse_rl_done {
            return self.parse_response_line_helper(line_len);
        }

        if !self.parse_rh_done {
            return self.parse_response_header(line_len, ignore_invalid_header);
        }

        unreachable!("response header already parse done");
    }

    fn parse_response_line_helper(&mut self, line_len: usize) -> Result<()> {
        let line_slice =
            &self.last_buf.as_ref().unwrap().read_slice()[0..line_len];

        let mut off: usize = 0;
        let end = line_len;

        // ignore resp version.
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v == b' ' {
                off = i;
                break;
            }
        }

        if off == 0 {
            return Error::e_explain(
                ErrorType::InvalidResponseLine,
                "invalid response version",
            );
        }

        // parse space before status code
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v != b' ' {
                off = i;
                break;
            }
        }

        let mut status_code = &line_slice[0..0];
        let status_code_start = off;
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if !v.is_ascii_digit() {
                off = i;
                status_code = &line_slice[status_code_start..i];
                break;
            }
        }

        let i: u16 = String::from_utf8_lossy(status_code).parse().if_err(
            ErrorType::InvalidResponseLine,
            "invalid response status code",
        )?;
        if !(100..=999).contains(&i) {
            return Error::e_explain(
                ErrorType::InvalidResponseLine,
                "invalid response status code",
            );
        }

        self.resp.status_code = i;

        // parse space before status reason.
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v != b' ' {
                off = i;
                break;
            }
        }

        let reason_end = end - 2;
        if off < reason_end {
            let reason = line_slice[off..reason_end].trim_ascii();
            self.resp.status_reason =
                String::from_utf8_lossy(reason).to_string();
        }

        self.parse_rl_done = true;

        Ok(())
    }

    fn parse_response_header(
        &mut self,
        line_len: usize,
        ignore_invalid_header: bool,
    ) -> Result<()> {
        let line_slice =
            &self.last_buf.as_ref().unwrap().read_slice()[0..line_len];

        match parse_http_header(line_slice) {
            Ok(kv) => {
                if kv.0.is_empty() {
                    self.parse_rh_done = true;
                    return Ok(());
                }

                let header = HttpHeader::with_kv(
                    Bytes::copy_from_slice(kv.0),
                    Bytes::copy_from_slice(kv.1),
                );

                self.process_header(&header)?;
                self.resp.headers.append_header(header);

                Ok(())
            }

            Err(e) => {
                if ignore_invalid_header {
                    self.invalid_headers += 1;
                    return Ok(());
                }

                Err(e)
            }
        }
    }

    fn process_header(&mut self, header: &HttpHeader) -> Result<()> {
        let header_name = header.get_name();

        match header_name.len() {
            7 => {
                if header_name.eq_ignore_ascii_case(HeaderName::TRAILER) {
                    let iter = header.get_value().split(|c| *c == b',');
                    for mut v in iter {
                        v = v.trim_ascii();
                        if !v.is_empty() {
                            self.resp_expect_trailers += 1;
                        }
                    }
                    return Ok(());
                }

                Ok(())
            }

            10 => {
                if header_name.eq_ignore_ascii_case(HeaderName::CONNECTION) {
                    if header.value_contain(HeaderValue::CHUNKED) {
                        self.want_close = true;
                        return Ok(());
                    }
                    return Ok(());
                }

                Ok(())
            }

            14 => {
                if header_name.eq_ignore_ascii_case(HeaderName::CONTENT_LENGTH)
                {
                    let value = header.get_value();
                    let res: usize =
                        String::from_utf8_lossy(value).parse().if_err(
                            ErrorType::InvalidHttpHeader,
                            "invalid content-length header",
                        )?;
                    self.resp_content_length = Some(res);
                    return Ok(());
                }

                Ok(())
            }

            17 => {
                if header_name
                    .eq_ignore_ascii_case(HeaderName::TRANSFER_ENCODING)
                {
                    if header.value_contain(HeaderValue::CHUNKED) {
                        self.resp_chunked_transfer = true;
                    }
                    return Ok(());
                }

                Ok(())
            }

            _ => {
                // nothing.
                Ok(())
            }
        }
    }

    fn process_response(&mut self) -> Result<()> {
        if self.resp.status_code < 200 || self.resp.status_code > 399 {
            self.want_close = true;
        }

        if !http_body_allow_status(self.resp.status_code) {
            self.resp_no_body = true;
        }

        self.init_body_reader();

        Ok(())
    }

    fn init_body_reader(&mut self) {
        let last_buf = self.last_buf.take();

        if self.resp.status_code == StatusCode::HTTP_SWITCHING_PROTOCOLS {
            self.resp_expect_trailers = 0;
            self.want_close = true;
            self.body_reader = Some(BodyReader::with_until_closed_transfer(
                last_buf,
                self.body_buf_size,
                self.body_read_timeout_ms,
            ));
            return;
        }

        if self.resp_no_body {
            self.resp_expect_trailers = 0;
            self.want_close = true;
            self.body_reader = Some(BodyReader::with_no_body(last_buf));
            return;
        }

        if self.resp_chunked_transfer {
            self.body_reader = Some(BodyReader::with_chunked_transfer(
                last_buf,
                self.body_buf_size,
                self.body_read_timeout_ms,
                self.resp_expect_trailers,
            ));
            return;
        }

        if let Some(cl) = self.resp_content_length {
            self.resp_expect_trailers = 0;
            self.body_reader = Some(BodyReader::with_cl_transfer(
                last_buf,
                cl,
                self.body_buf_size,
                self.body_read_timeout_ms,
            ));
            return;
        }

        self.resp_expect_trailers = 0;
        self.want_close = true;
        self.body_reader = Some(BodyReader::with_until_closed_transfer(
            last_buf,
            self.body_buf_size,
            self.body_read_timeout_ms,
        ));
    }
}
